package com.trusfort.redis;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.*;
import redis.clients.jedis.Protocol.Command;
import redis.clients.jedis.params.ScanParams;
import redis.clients.jedis.resps.ScanResult;
import redis.clients.jedis.util.JedisClusterCRC16;
import redis.clients.jedis.util.SafeEncoder;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * Redis Cluster服务类
 * 
 * @author DangT
 * @date 2017年5月11日 下午2:39:43
 * @version V1.0
 */
public class RedisClusterService implements RedisService {
	
	private static List<String> CLUSTER_HOST =  Arrays.asList("192.168.1.170:7001,192.168.1.170:7002,192.168.1.170:7003,192.168.1.170:7004,192.168.1.170:7005,192.168.1.170:7006".split(",")) ;
	
	
	private static int MAX_REDIRECTIONS = 5;
	
	 // 可用连接实例的最大数目
    private static int MAX_ACTIVE = 300;
    // 一个pool最多有多少个状态为idle(空闲的)的Jedis实例
    private static int MAX_IDLE = 30;
    // 等待可用连接的最大时间（单位：毫秒）
    private static int MAX_WAIT = 10000;
    // 在borrow一个Jedis实例时，是否提前进行validate操作
    private static boolean TEST_ON_BORROW = false;

    private static String PASSWORD = null;
    
    // 超时时间（单位：毫秒）
    private static int TIME_OUT = 3000;
	
	private static Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
	private static JedisCluster cluster;
	

	private static RedisClusterService service;
	
	 private static volatile RedisClusterService instance;
	 private static GenericObjectPoolConfig poolConfig;
	
 
	private static String HOSTANDPORT;
	

	private RedisClusterService() {
	}

	
	// 私有构造方法
    private RedisClusterService(String hostAndPort, String password) throws Exception {
        initializeCluster(hostAndPort, password);
    }
    
    // 双重检查锁单例
    public static RedisClusterService getInstance(String hostAndPort, String password) throws Exception {
//        if (instance == null) {
//            synchronized (RedisClusterService.class) {
//                if (instance == null) {
//                    instance = new RedisClusterService(hostAndPort, password);
//                }
//            }
//        }
    	instance = new RedisClusterService(hostAndPort, password);
        return instance;
    }
	
    
    // 初始化集群连接
    private void initializeCluster(String hostAndPort, String password) throws Exception {
        Set<HostAndPort> jedisClusterNodes = new HashSet<>();
        List<String> hosts = Arrays.asList(hostAndPort.split(","));
        
        for (String host : hosts) {
            String[] split = host.split(":");
            jedisClusterNodes.add(new HostAndPort(split[0], Integer.parseInt(split[1])));
        }
        
        poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(300);
        poolConfig.setMaxIdle(30);
        poolConfig.setMaxWaitMillis(10000);
        poolConfig.setTestOnBorrow(false);
        
        if (password == null || password.trim().isEmpty()) {
            cluster = new JedisCluster(jedisClusterNodes, 3000, 3000, 5, poolConfig);
        } else {
            cluster = new JedisCluster(jedisClusterNodes, 3000, 3000, 5, password, poolConfig);
        }
        
        System.out.println("Redis Cluster initialized: " + cluster.ping());
    }
    
	/**
	 * 初始化Redis
	 * @return 
	 *
	 */
	public static RedisClusterService init(String hostAndPort,String passwd) throws Exception {
		
		try {
//			if (service == null) {
				service = new RedisClusterService();
				
				HOSTANDPORT = hostAndPort; 
				PASSWORD = passwd;
				
				CLUSTER_HOST = Arrays.asList(HOSTANDPORT.split(",")) ;
				
				for (String host : CLUSTER_HOST) {
					String[] split = host.split(":");
					jedisClusterNodes.add(new HostAndPort(split[0], Integer.parseInt(split[1])));
				}
				

				GenericObjectPoolConfig config = new GenericObjectPoolConfig();
				config.setMaxTotal(MAX_ACTIVE);
				config.setMaxIdle(MAX_IDLE);
				config.setMaxWaitMillis(MAX_WAIT);
				config.setTestOnBorrow(TEST_ON_BORROW);

				if(passwd == null) {
					cluster = new JedisCluster(jedisClusterNodes, TIME_OUT, TIME_OUT, MAX_REDIRECTIONS,  config);	
				}else {
					cluster = new JedisCluster(jedisClusterNodes, TIME_OUT, TIME_OUT, MAX_REDIRECTIONS, passwd, config);
				}
				
				 System.out.println(cluster.ping());;
				
//			}
		} catch (Exception e) {
//			logger.error("Initialize redis error!", e);
			throw e;
		}
		
		return service;
		
	}
	
	     
	public void set(String key, String value) throws Exception {

		cluster.set(key, value);

	}

	public void set(Integer dbIndex, String key, String value) throws Exception {

		set(key, value);
		
	}

	public void set(String key, String value, int seconds) throws Exception {

		cluster.set(key, value);
		cluster.expire(key, seconds);

	}

	public void set(Integer dbIndex, String key, String value, int seconds) throws Exception {

		set(key, value, seconds);
		
	}
	
//	public ScanResult<String> scan(DBIndex dbIndex, final String key,String cursor,Integer count) {
//		
//		Jedis jedis = null;
//		try {
//			ScanParams scanParams=new ScanParams();
//			scanParams.count(count);
//			//使用sscan命令获取500条数据，使用cursor游标记录位置，下次循环使用
//			return cluster.scan(cursor, scanParams);
//
//		} catch (Exception e) {
//			e.printStackTrace();
////			log.error("{}", ERR_MSG, e);
//			return null;
//		} finally {
//			RedisConnectionPool.getInstance().returnResource(jedis);
//		}
//	}
	
    /**
     * 扫描单个节点的key
     */
    private Set<String> scanNode(Jedis jedis, String pattern,Integer count) {
        Set<String> keys = new HashSet<>();
        String cursor = ScanParams.SCAN_POINTER_START;
        ScanParams scanParams = new ScanParams().match(pattern).count(count);
        
        do {
            ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
            keys.addAll(scanResult.getResult());
            cursor = scanResult.getCursor();
        } while (!cursor.equals(ScanParams.SCAN_POINTER_START));
        
        return keys;
    }
	
    
    private void authenticateIfRequired(Jedis jedis) {
        try {
            // 如果配置了密码，进行认证
            String password = "123456";
            if (password != null && !password.trim().isEmpty()) {
                jedis.auth(password);
            }
        } catch (Exception e) {
//            logger.warn("认证失败: {}", e.getMessage());
            // 在某些情况下可能已经认证过，继续执行
        }
    }
    
 // 使用类成员变量保存每个节点的扫描状态
    private Map<String, String> nodeCursors = new ConcurrentHashMap<>();

    public ScanResult<String> scan(Integer dbIndex, String key, String cursor, Integer count) {
		
        Set<String> allKeys = new HashSet<>();
        Map<String, ConnectionPool> nodeMap = cluster.getClusterNodes();
        
        // 初始化扫描
        if ("0".equals(cursor)) {
            nodeCursors.clear();
            nodeMap.keySet().forEach(node -> nodeCursors.put(node, "0"));
        }
        
        boolean allFinished = true;
        
        for (Map.Entry<String, ConnectionPool> entry : nodeMap.entrySet()) {
            String nodeId = entry.getKey();
            String nodeCursor = nodeCursors.get(nodeId);
            
            if ("0".equals(nodeCursor) && !"0".equals(cursor)) {
                continue; // 该节点已经扫描完成
            }
            
            try (Connection connection = entry.getValue().getResource()) {
            	
            	/*
            	 改为下面方式，此处代码会要输入授权密码
                try (Jedis jedis = new Jedis(connection)) {
                	
//                	authenticateIfRequired(jedis);
                	
                	 // 只处理从节点（反转判断条件）
                    if (!isMasterNode(jedis)) {
                        continue;
                    }
                	
                    ScanResult<String> scanResult = jedis.scan(
                        nodeCursor,
                        new ScanParams().match(key).count(count)
                    );
                    
                    allKeys.addAll(scanResult.getResult());
                    nodeCursors.put(nodeId, scanResult.getCursor());
                    
                    if (!"0".equals(scanResult.getCursor())) {
                        allFinished = false;
                    }
                }
                */
                
                connection.sendCommand(Command.SCAN, nodeCursor, "MATCH", key, "COUNT", String.valueOf(count));
                
                List<Object> response = connection.getObjectMultiBulkReply();
                String newCursor = SafeEncoder.encode((byte[]) response.get(0));
                List<byte[]> rawKeys = (List<byte[]>) response.get(1);
                
                // 转换结果
                List<String> keys = rawKeys.stream()
                    .map(SafeEncoder::encode)
                    .collect(Collectors.toList());
                
                allKeys.addAll(keys);
                nodeCursors.put(nodeId, newCursor);
                
                if (!"0".equals(newCursor)) {
                    allFinished = false;
                }
                
            }
        }
        
        String nextCursor = allFinished ? "0" : "1"; // 使用非0值表示还有更多数据
        return new ScanResult<>(nextCursor, new ArrayList<>(allKeys));
    }

	private boolean isMasterNode(Jedis jedis) {
	    String info = jedis.info("Replication");
	    //return info.contains("role:master");
	    return info.contains("role:master");
	}
	

	@Override
	public void close() {
		
		if (cluster != null) {
			try {
				cluster.close();
			} catch (Exception e) {
//                logger.warn("", e);
			}
		}

	}

	@Override
	public long ttl(String key) throws Exception {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public void set(List<String> list, int seconds) throws Exception {
		if (list == null || list.isEmpty()) {
	        return;
	    }
	    
	    // 使用并行流提高效率
	    long successCount = list.parallelStream()
	        .map(key -> {
	            try {
	            	
	            	String [] kv = key.split(";");
	                cluster.set(kv[0], kv[1]);
	                return true;
	            } catch (Exception e) {
	                System.err.println("add key failed: " + key + ", error: " + e.getMessage());
	                return false;
	            }
	        })
	        .filter(success -> success)
	        .count();
	    
	    System.out.println("Successfully added " + successCount + " out of " + list.size() + " keys");
		
	}
//
//	private void authenticateCluster() {
//	    // 获取集群所有节点的连接并执行认证
//		Map<String, ConnectionPool> clusterNodes = cluster.getClusterNodes();
//	    for (Map.Entry<String, ConnectionPool> entry : clusterNodes.entrySet()) {
//	    	
//	         try (Connection connection = entry.getValue().getResource()) {
//	                try (Jedis jedis = new Jedis(connection)) {
//	                	jedis.auth("123456");
//	                }
//	            }
//	    }
//	}
	
	@Override
	public void del(Integer dbIndex, List<String> list) throws Exception {
		if (list == null || list.isEmpty()) {
	        return;
	    }
	  
		// 然后执行并行删除操作
		long successCount = list.parallelStream()
		    .map(key -> {
		        try {
		            cluster.del(key);
		            return true;
		        } catch (Exception e) {
		            System.err.println("Delete key failed: " + key + ", error: " + e.getMessage());
		            return false;
		        }
		    })
		    .filter(success -> success)
		    .count();
	    System.out.println("Successfully deleted " + successCount + " out of " + list.size() + " keys");
	}

}
